package com.gpj.learn.rocketmq.radio;

import com.gpj.learn.rocketmq.CommonConstants;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

/**
 * @className RadioConsumer
 * @description
 *   广播消息的消息消费者(需要指定消息消费模式)
 *   需要启动两个实例,来观察效果
 * @author GPJ
 * @date 2021/4/24 16:30
 * @version 1.0
 **/
@Slf4j
public class RadioConsumer {

	public static void main(String[] args) {
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("RadioMessageConsumerGroup");
		consumer.setNamesrvAddr(CommonConstants.ROCKETMQ_NAMESERVER_ADDR);

		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);

		consumer.setMessageModel(MessageModel.BROADCASTING); // 设置广播消费模式

		try {
			consumer.subscribe(CommonConstants.DEFAULT_TOPIC_NAME, "*");
			consumer.registerMessageListener((MessageListenerConcurrently) (list, context) -> {
				for (MessageExt message : list) {
					log.info("接收到消息:{}", message);
				}
				return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
			});
			consumer.start();
			log.info("consumer starting...");
		} catch (MQClientException e) {
			e.printStackTrace();
		}
	}

}
